package com.ucode.springboot.starter.canal.client.transfer;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

import org.springframework.util.CollectionUtils;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.ucode.springboot.starter.canal.annotation.ListenPoint;
import com.ucode.springboot.starter.canal.client.CanalMsg;
import com.ucode.springboot.starter.canal.client.ListenerPoint;
import com.ucode.springboot.starter.canal.config.CanalConfig;
import com.ucode.springboot.starter.canal.event.CanalEventListener;

/**
 * 消息处理转换类
 * @author: liliang
 * @date: 2020年2月27日 下午4:52:09
 */
public abstract class AbstractBasicMessageTransponder extends AbstractMessageTransponder{

    public AbstractBasicMessageTransponder(CanalConnector connector, Map.Entry<String, CanalConfig.Instance> config, List<CanalEventListener> listeners, List<ListenerPoint> annoListeners) {
        super(connector, config, listeners, annoListeners);
    }

    @Override
    protected void distributeEvent(Message message) {
        
        //获取操作实体
        List<CanalEntry.Entry> entries = message.getEntries();
        //遍历实体
        for (CanalEntry.Entry entry : entries) {
            //忽略实体类的类型
            List<CanalEntry.EntryType> ignoreEntryTypes = getIgnoreEntryTypes();
            if (ignoreEntryTypes != null
                    && ignoreEntryTypes.stream().anyMatch(t -> entry.getEntryType() == t)) {
                continue;
            }
            //canal 改变信息
            CanalEntry.RowChange rowChange;
            try {
                //获取信息改变
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                
            } catch (Exception e) {
                throw new CanalClientException("错误 ##转换错误 , 数据信息:" + entry.toString(),
                        e);
            }
            
            distributeByAnnotation(destination,
                    entry.getHeader().getSchemaName(),
                    entry.getHeader().getTableName(), rowChange);
            distributeByImpl(destination,
                    entry.getHeader().getSchemaName(),
                    entry.getHeader().getTableName(), rowChange);
            
        }
    }
    
    /**
     * 处理注解方式的 canal 监听器
     * @date 2020年2月27日
     * @param destination canal instances key
     * @param schemaName 实例名称
     * @param tableName 表名称
     * @param rowChange 数据
     */
    protected void distributeByAnnotation(String destination, String schemaName, String tableName, CanalEntry.RowChange rowChange) {
        
        //对注解的监听器进行事件委托
        if (!CollectionUtils.isEmpty(annoListeners)) {
            annoListeners.forEach(point -> point
                    .getInvokeMap()
                    .entrySet()
                    .stream()
                    .filter(getAnnotationFilter(destination, schemaName, tableName, rowChange.getEventType()))
                    .forEach(entry -> {
                        Method method = entry.getKey();
                        method.setAccessible(true);
                        try {
                            CanalMsg canalMsg = new CanalMsg();
                            canalMsg.setDestination(destination);
                            canalMsg.setSchemaName(schemaName);
                            canalMsg.setTableName(tableName);
                            
                            Object[] args = getInvokeArgs(method, canalMsg, rowChange);
                            method.invoke(point.getTarget(), args);
                        } catch (Exception e) {
                            log.error(e,"{}: 委托 canal 监听器发生错误! 错误类:{}, 方法名:{}",
                                    Thread.currentThread().getName(),
                                    point.getTarget().getClass().getName(), method.getName());
                        }
                    }));
        }
    }
    

    /**
     * 处理监听信息
     * @date 2020年2月27日
     * @param destination canal instances key
     * @param schemaName 数据库实例
     * @param tableName 表名称
     * @param rowChange 处理的数据
     */
    protected void distributeByImpl(String destination, String schemaName, String tableName, CanalEntry.RowChange rowChange) {
        if (listeners != null) {
            for (CanalEventListener listener : listeners) {
                listener.onEvent(destination, schemaName, tableName, rowChange);
            }
        }
    }
    
    
    /**
     *  忽略实体类的类型
     * @date 2020年2月27日
     * @return
     */
    protected List<CanalEntry.EntryType> getIgnoreEntryTypes() {
        return Collections.emptyList();
    }
    
    /**
     * 断言注解方式的监听过滤规则
     * @date 2020年2月27日
     * @param destination canal instances key
     * @param schemaName 数据库实例
     * @param tableName 表名称
     * @param eventType 事件类型
     * @return
     */
    protected abstract Predicate<Map.Entry<Method, ListenPoint>> getAnnotationFilter(String destination, String schemaName, String tableName, CanalEntry.EventType eventType);
    
    /**
     * 获取处理的参数
     * @date 2020年2月27日
     * @param method 委托处理的方法
     * @param canalMsg  事件节点
     * @param rowChange 处理的数据
     * @return
     */
    protected abstract Object[] getInvokeArgs(Method method, CanalMsg canalMsg, CanalEntry.RowChange rowChange);
}
